package top.joylee.example.grpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.joylee.example.grpc.route.*;

import java.io.IOException;
import java.net.URL;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

//服务端程序
public class RouteGuideServer {
    private final static Logger log = LoggerFactory.getLogger(RouteGuideServer.class);

    private final int port;
    private final Server server;

    //构造函数
    //指定服务端监听端口号
    //同时，通过工具将json文件中定义好的点位特征值（Feature）加载到内存中来
    //RouteGuideUtil用来模拟数据库的查询操作，详见RouteGuideUtil代码
    public RouteGuideServer(int port) throws IOException {
        this(port, RouteGuideUtil.getDefaultFeaturesFile());
    }

    private RouteGuideServer(int port, URL featureFile) throws IOException {
        this(ServerBuilder.forPort(port), port, RouteGuideUtil.parseFeatures(featureFile));
    }

    RouteGuideServer(ServerBuilder<?> serverBuilder, int port, Collection<Feature> features) {
        this.port = port;
        //使用ServerBuilder构建service
        //其中RouteGuideService类是继承的RouteGuideGrpc.RouteGuideImplBase
        //在该类中实现了.proto的四个方法
        server = serverBuilder.addService(new RouteGuideService(features))
                .build();
    }

    public static void main(String[] args) throws Exception {
        //第一步：创建一个服务端，指定监听端口
        RouteGuideServer server = new RouteGuideServer(18080);
        //第二步：启动服务端
        server.start();
        //第三步：阻塞，直到关闭服务
        server.blockUntilShutdown();
    }

    public void start() throws IOException {
        //如果server为空，证明创建server对象失败，无法启动
        if (Objects.isNull(server)) {
            log.error("route guide server start fail...");
            return;
        }
        //启动server
        server.start();
        log.info("Server started,listening {}......", port);

        //启动成功后，添加监听jvm关闭的钩子函数
        //如果JVM关闭，则关闭此服务端
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.warn("*** shutting down gRPC server  since JVM shutdown ***");
            try {
                stop();
            } catch (InterruptedException e) {
                log.error("await stop error,interrupted", e);
            }
            log.info("**** server shut down ,good bye ***");
        }));
    }

    public void stop() throws InterruptedException {
        if (server != null) {
            server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
        }
    }

    private void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }

    //实现.proto文件中定义的rpc服务
    private static class RouteGuideService extends RouteGuideGrpc.RouteGuideImplBase {
        //用来保存所有“数据库中”存储的特征值
        //本示例中，会在new RouteGuideService 对象时通过工具，将json文件中的值加载进来
        private final Collection<Feature> features;

        //点位值对应的RouteNote对象列表映射
        private final ConcurrentMap<Point, List<RouteNote>> routeNotes =
                new ConcurrentHashMap<>();

        public RouteGuideService(Collection<Feature> features) {
            this.features = features;
        }


        /**
         * 查询point是否在“数据库”中有点位值
         *
         * @param request          point信息
         * @param responseObserver 返回Feature，执行完业务逻辑后，直接completed
         */
        @Override
        public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
            responseObserver.onNext(checkFeature(request));
            responseObserver.onCompleted();
        }

        /**
         * 获取指定返回内的特征点位值
         *
         * @param request          指定要查询的范围
         * @param responseObserver 服务端以流的方式响应客户端。
         */
        @Override
        public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
            //左边界，取决于最小经度
            int left = Math.min(request.getLo().getLongitude(), request.getHi().getLongitude());
            //右边界，取决于最大经度
            int right = Math.max(request.getLo().getLongitude(), request.getHi().getLongitude());
            //上边界，取决于最大纬度
            int top = Math.max(request.getLo().getLatitude(), request.getHi().getLatitude());
            //下边界，取决于最小纬度
            int bottom = Math.min(request.getLo().getLatitude(), request.getHi().getLatitude());

            // 遍历“数据库”中所有的特征点位
            // 假设：这个过程会比较漫长,如果一次性的返回所有，则会等待很长时间才响应
            //      但是这里是以流的方式响应客户端，也就是在for循环中，找到一个就响应一个给客户端
            //      直到遍历结束才completed，结束服务端的响应
            for (Feature feature : features) {
                if (!RouteGuideUtil.exists(feature)) {
                    continue;
                }
                int lat = feature.getLocation().getLatitude();
                int lon = feature.getLocation().getLongitude();
                if (lon >= left && lon <= right && lat >= bottom && lat <= top) {
                    responseObserver.onNext(feature);
                }
            }
            responseObserver.onCompleted();
        }


        /**
         * 客户端流式调用
         *
         * @param responseObserver 服务端以流的方式将point列表发送给服务端
         * @return 服务端将提取流中的消息，并将收到的所有消息统计后，返回给客户端（服务端一次响应）
         */
        @Override
        public StreamObserver<Point> recordRoute(StreamObserver<RouteSummary> responseObserver) {
            //【特别注意】：由于客户端是以流的方式将多条消息（“消息列表”）请求过来，所以这里要建一个基于流请求的StreamObserver，去处理请求流
            return new StreamObserver<Point>() {
                int pointCount;
                int featureCount;
                int distance;
                Point previous;
                final long startTime = System.nanoTime();

                //处理请求流，来一个处理一个
                @Override
                public void onNext(Point point) {
                    pointCount++;
                    if (RouteGuideUtil.exists(checkFeature(point))) {
                        featureCount++;
                    }
                    if (Objects.nonNull(previous)) {
                        distance += calcDistance(previous, point);
                    }
                    previous = point;
                }

                @Override
                public void onError(Throwable throwable) {
                    log.error("record route cancel", throwable);
                }

                //请求流处理完成，则响应
                //只用响应单个请求体
                //至于这里的onCompleted方法，此方法是当客户端流完成传输后执行的方法。
                @Override
                public void onCompleted() {
                    long seconds = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - startTime);
                    responseObserver.onNext(
                            RouteSummary.newBuilder()
                                    .setPointCount(pointCount)
                                    .setDistance(distance)
                                    .setFeatureCount(featureCount)
                                    .setElapsedTime((int) seconds)
                                    .build());
                    //此处是结束服务端响应的方法。
                    responseObserver.onCompleted();
                }
            };
        }

        /**
         * 双向流式调用
         * 这里可以演示，服务端并不一定要等待客户端消息完成再响应
         * 而是来一条响应一条，可以看到客户端以边在发送消息，一边在接受服务端的响应
         * 两条流是独立的
         *
         * @param responseObserver 请求流
         * @return 可以这么认为，客户端A、客户端B都可以到某个点位，并做个记录。如果下一个人来到此点位，则返回曾经到过此处的Note信息。如果从来没有，则不返回。
         */
        @Override
        public StreamObserver<RouteNote> routeChat(StreamObserver<RouteNote> responseObserver) {
            return new StreamObserver<RouteNote>() {
                @Override
                public void onNext(RouteNote note) {
                    List<RouteNote> notes = getOrCreateNotes(note.getLocation());
                    for (RouteNote prevNode : notes.toArray(new RouteNote[0])) {
                        responseObserver.onNext(prevNode);
                    }
                    notes.add(note);
                }

                @Override
                public void onError(Throwable throwable) {
                    log.error("routeChat  cancel", throwable);
                }

                @Override
                public void onCompleted() {
                    responseObserver.onCompleted();
                }
            };
        }


        //检查点位是否在“数据库”特征值中
        //如果存在，则返回由point + name组成的特征点位信息
        //否则，返回由point + name=“” 的特征点位信息
        private Feature checkFeature(Point location) {
            for (Feature feature : features) {
                if (Objects.equals(feature.getLocation().getLatitude(), location.getLatitude())
                        && Objects.equals(feature.getLocation().getLongitude(), location.getLongitude())) {
                    return feature;
                }
            }
            return Feature.newBuilder().setName("").setLocation(location).build();
        }

        private static int calcDistance(Point start, Point end) {
            int r = 6371000;
            double lat1 = Math.toRadians(RouteGuideUtil.getLatitude(start));
            double lat2 = Math.toRadians(RouteGuideUtil.getLatitude(end));
            double lon1 = Math.toRadians(RouteGuideUtil.getLongitude(start));
            double lon2 = Math.toRadians(RouteGuideUtil.getLongitude(end));
            double deltaLat = lat2 - lat1;
            double deltaLon = lon2 - lon1;
            double a = Math.sin(deltaLat / 2) * Math.sin(deltaLat / 2)
                    + Math.cos(lat1) * Math.cos(lat2) * Math.sin(deltaLon / 2) * Math.sin(deltaLon / 2);
            double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));
            return (int) (r * c);
        }

        private List<RouteNote> getOrCreateNotes(Point location) {
            List<RouteNote> notes = Collections.synchronizedList(new ArrayList<>());
            List<RouteNote> prevNotes = routeNotes.putIfAbsent(location, notes);
            //如果此point之前存在，则返回之前所有到过此处的RouteNote信息
            return CollectionUtils.isEmpty(prevNotes) ? notes : prevNotes;
        }
    }

}
